package com.pie4cloud.pie.bus.controller;

import com.pie4cloud.pie.bus.api.consumer.MetricCalcService;
import com.pie4cloud.pie.bus.api.dto.ConsumerDto;
import com.pie4cloud.pie.bus.api.dto.MessageDto;
import com.pie4cloud.pie.bus.service.KafkaSendService;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.common.security.annotation.Inner;
import io.swagger.annotations.Api;
import lombok.AllArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.validation.Valid;
import java.util.ArrayList;
import java.util.List;

@RestController
@AllArgsConstructor
@RequestMapping("/kfk")
@Api(value = "kfk", tags = "kafka消息管理")
public class KafkaSendController {
	@Autowired
	private KafkaSendService kafkaSendService;
	@Autowired
	private MetricCalcService metricCalcService;

	@PostMapping("/sendKafka")
	public R sendKafka(@Valid @RequestBody MessageDto messageDto) {
        List<MessageDto> messageDtos=new ArrayList<>();
		messageDtos.add(messageDto);
		metricCalcService.submit(messageDtos);
		return R.ok();
	}

	@Inner
	@PostMapping("/sendKafkas")
	public R sendKafkas(@RequestBody List<MessageDto> messageDtos) {
		R r=kafkaSendService.sendKafkas(messageDtos);
		return r;
	}
	@PostMapping("/sendConsumer")
	public R  sendConsumer(@Valid @RequestBody ConsumerDto consumerDto) {
		R r=kafkaSendService.sendConsumer(consumerDto);
		return r;
	}
}
